package com.glodon.paas.job.agent.worker;

import static com.glodon.paas.job.JobConstants.*;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.glodon.paas.job.agent.cli.Cli;
import com.glodon.paas.job.exception.JsonException;
import com.glodon.paas.job.model.Task;
import com.glodon.paas.job.util.JsonUtil;

@SuppressWarnings({ "rawtypes", "unchecked" })
public class DefaultWorker implements Worker {
    // default timeout 10 hours
    private static final int DEFAULT_TIMEOUT = 10 * 60 * 60 * 1000;
    private static final String CLI_PARAM_TASKID = "taskid";
    private static final Logger logger = LoggerFactory.getLogger(DefaultWorker.class);
    private Cli cli;
    private Timer timeoutTimer = new Timer(true);
    private boolean timeout;
    private Process exec;

    private WorkerTimeoutTask workerTimeoutTask;
    private Caller caller;
    private String cmd;
    private String[] envp;

    public void setCli(Cli cli) {
        this.cli = cli;
    }

    @Override
    public void cancel() throws WorkerException {
        try {
            timeout = true;
            exec.getInputStream().close();
            exec.getErrorStream().close();
            exec.getOutputStream().close();
            exec.destroy();
        } catch (IOException e) {
            throw new WorkerException("cancel worker failure", e);
        }
    }

    @Override
    public Map work(Task task) throws WorkerException {
        String type = task.getTaskType();
        Map params = task.getValues();
        try {
            setTimeout(DEFAULT_TIMEOUT);
            params.put(CLI_PARAM_TASKID, task.getTaskId());
            logger.info("process start, type: {}, params: {}", type, params);
            Runtime runtime = Runtime.getRuntime();
            String cmd = this.cmd + " " + cli.generate(params);
            logger.info("worker execute command: {}", cmd);
            exec = runtime.exec(cmd, envp);
            FutureTask<String> futureTask = new FutureTask<String>(new Callable<String>() {
                @Override
                public String call() throws Exception {
                    return getErrorMsg();
                }
            });
            new Thread(futureTask).start();
            processStdInfo();
            String errorMsg = futureTask.get();
            int exitValue = exec.waitFor();
            logger.info("process end, type: {}, params: {}, exit value: {}", new Object[] { type, params, exitValue });
            // when timeout
            if (timeout)
                throw new WorkerException(TIMEOUT_MSG);
            if (exitValue != 0)
                throw new WorkerException(errorMsg);
            Map result = new HashMap();
            result.put(ERROR_MSG, errorMsg);
            // TODO for test, add input params to result
            result.putAll(params);
            return result;
        } catch (IOException e) {
            throw new WorkerException(e);
        } catch (InterruptedException e) {
            throw new WorkerException(e);
        } catch (ExecutionException e) {
            throw new WorkerException(e);
        } finally {
            cancelTimeTask();
        }
    }

    private void processStdInfo() throws IOException {
        BufferedReader std = new BufferedReader(new InputStreamReader(exec.getInputStream()));
        String line;
        while ((line = std.readLine()) != null) {
            logger.info("std output: {}", line);
            process(line);
        }
        std.close();
    }

    private String getErrorMsg() {
        try {
            InputStream errorStream = exec.getErrorStream();
            String errorMsg = IOUtils.toString(errorStream, "UTF-8");
            errorStream.close();
            return errorMsg;
        } catch (Exception e) {
            return e.getMessage();
        }
    }

    // process std cmmand
    private void process(String line) {
        logger.info("process line: {}", line);
        if (StringUtils.isBlank(line) || !line.startsWith("{"))
            return;
        try {
            Map result = JsonUtil.jsonToMap(line);
            processSendProgress(result);
            processSetTimeout(result);
            processUnsetTimeout(result);
        } catch (JsonException e) {
            logger.warn("std output is not json: {}", line, e);
        }
    }

    private void processUnsetTimeout(Map result) {
        Object unsetTimeout = result.get(UNSET_TIMEOUT);
        if (unsetTimeout != null) {
            cancelTimeTask();
        }
    }

    private void processSetTimeout(Map result) {
        Object tmp = result.get(SET_TIMEOUT);
        if (tmp == null)
            return;
        if (tmp instanceof Number) {
            Number num = (Number) tmp;
            int setTimeout = num.intValue();
            setTimeout(setTimeout);
        }
    }

    private void setTimeout(int timeout) {
        logger.info("set timeout: {}", timeout);
        cancelTimeTask();
        workerTimeoutTask = new WorkerTimeoutTask(this,cmd);
        timeoutTimer.schedule(workerTimeoutTask, timeout);
    }

    private void cancelTimeTask() {
        if (workerTimeoutTask != null) {
            workerTimeoutTask.cancel();
            timeoutTimer.purge();
        }
    }

    private void processSendProgress(Map result) {
        Object progress = result.get(TASK_PROGRESS);
        if (progress != null) {
            int p = ((Number) progress).intValue();
            String message = (String) result.get(TASK_MESSAGE);
            caller.sendProgress(p, message);
        }
    }

    public void setCaller(Caller caller) {
        this.caller = caller;
    }

    public void setCmd(String cmd) {
        this.cmd = cmd;
    }

    public void setEnvp(String[] envp) {
        this.envp = envp;
    }
}
